﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using UtilityToolsCollect.ObjectsLibrary.ProcessManage;

namespace UtilityToolsCollect.ObjectsLibrary.TaskManage
{
    /// <summary>
    /// 只要有释放或者取消操作，该类型的实例作废，是一次性实例，不可在取消操作后继续使用，内部状态不可恢复
    /// </summary>
    public class LimitedParallelPerformanceTaskManage : IDisposable
    {
        public abstract class TaskWrapperBase
        {
            protected Task m_task;
            protected Task? UnwrapTask;
            public int TaskID => m_task.Id;
            public bool IsCanceled => UnwrapTask is not null ? UnwrapTask.IsCanceled : m_task.IsCanceled;
            public CancellationTokenSource cancellationTokenSource { get; init; }
            public TaskWrapperBase(Task task, CancellationTokenSource cancellationTokenSource)
            {
                m_task = task;
                this.cancellationTokenSource = cancellationTokenSource;
            }
            public async Task ExecuteTask()
            {
                m_task.Start();
                if (UnwrapTask is not null)
                    await UnwrapTask.WaitAsync(cancellationTokenSource.Token);
                else
                    await m_task.WaitAsync(cancellationTokenSource.Token);
            }
            public void Cancel()
            {
                cancellationTokenSource.Cancel();
            }
        }
        public class TaskWrapper : TaskWrapperBase { public Task task => m_task; public TaskWrapper(Task task, CancellationTokenSource cancellationTokenSource) : base(task, cancellationTokenSource) { } }
        public class TaskWrapper<TResult> : TaskWrapperBase
        {
            public Task<TResult> task => (Task<TResult>)m_task;
            public TaskWrapper(Task<TResult> task, CancellationTokenSource cancellationTokenSource) : base(task, cancellationTokenSource) { }
        }
        public class TaskWrapperTaskResult<TResult> : TaskWrapper<Task<TResult>>
        {
            public TaskWrapper<Task<TResult>> TaskWrapper => this;
            public TaskWrapperTaskResult(Task<Task<TResult>> task, CancellationTokenSource cancellationTokenSource) : base(task, cancellationTokenSource) { UnwrapTask = task.Unwrap(); }
        }


        static LimitedParallelPerformanceTaskManage() {  }

        //核心字段

        public readonly int MaximumConcurrencyLevel;
        private ConcurrentQueue<TaskWrapperBase> TaskQueue = new();
        public IEnumerable<TaskWrapperBase> Queuings => TaskQueue;

        private ConcurrentDictionary<TaskWrapperBase, int> TaskQueueDic = new();

        private ConcurrentDictionary<TaskWrapperBase, int> RuningTasks = new();
        public IEnumerable<TaskWrapperBase> Runings => RuningTasks.Keys;

        private CancellationTokenSource? cancellationTokenSource;
        private CancellationToken cancellationToken;
        public bool IsActive => !disposedValue && !cancellationToken.IsCancellationRequested;

        //CPU占用侦测功能字段
        private System.Threading.PeriodicTimer? PollingTask;
        //private System.Threading.Timer? PollingTask;
        private CpuPerformanceCount? _cpuPerformanceCount;
        private float[]? _cpuPerformanceCountValues;
        private int CpuPerformanceCountIndex = 0;
        private float _CpuPerformanceLimit;
        public float CpuPerformanceAverage
        {
            get
            {
                IEnumerable<float>? temp = _cpuPerformanceCountValues?.Where(p => p > 0);
                if (temp?.Any() != true)
                    return -1;
                return temp.Average();
            }
        }

        private bool IsCpuLimit
        {
            get
            {
                if (_cpuPerformanceCountValues == null)
                    return false;
                AddCpuPerformanceCount();
                try
                {
                    float cpuAverageValue = CpuPerformanceAverage;
                    return (cpuAverageValue > 0 ? cpuAverageValue : float.MaxValue) > _CpuPerformanceLimit;
                }
                catch (InvalidOperationException)
                { return true; }
            }
        }
        private void AddCpuPerformanceCount()
        {
            if (_cpuPerformanceCount is null)
                return;
            float value = _cpuPerformanceCount.Value;
            if (value < 1)
                return;
            _cpuPerformanceCountValues![CpuPerformanceCountIndex] = value;
            CpuPerformanceCountIndex = (CpuPerformanceCountIndex + 1) % _cpuPerformanceCountValues.Length;
        }

        public LimitedParallelPerformanceTaskManage(int maximumConcurrencyLevel = 0, CancellationToken? token = null, bool EnableCpuPerformanceCount = false, float CpuPerformanceLimit = 60f, int PerformanceCountCacheSize = 20, int PollingInterval = 250)
        {
            MaximumConcurrencyLevel = maximumConcurrencyLevel > 0 ? maximumConcurrencyLevel : int.MaxValue;
            _CpuPerformanceLimit = CpuPerformanceLimit;
            if (token is not null)
                cancellationToken = token.Value;
            else
            {
                cancellationTokenSource = new();
                cancellationToken = cancellationTokenSource.Token;
            }
            if (!EnableCpuPerformanceCount)
                return;
            _cpuPerformanceCountValues = new float[PerformanceCountCacheSize];
            _cpuPerformanceCount = CpuPerformanceCount.Create(CpuPerformanceCount.PerformanceCountTypeEnum.InformationProcessorTime);
            _ = _cpuPerformanceCount.Value;
            PollingTask = new PeriodicTimer(TimeSpan.FromMilliseconds(PollingInterval));
            Task.Run(Polling);
        }

        public void QueueTask(TaskWrapperBase task)
        {
            if (!IsActive)
                return;
            TaskQueue.Enqueue(task);
            TryExecuteNextTask();
        }

        private void TryExecuteNextTask()
        {
            if (_cpuPerformanceCount is not null)
                return;
            while (IsActive && RuningTasks.Count < MaximumConcurrencyLevel && !IsCpuLimit && TaskQueue.TryDequeue(out TaskWrapperBase? TakeTask) && !TakeTask.IsCanceled)
                ExecuteTask(TakeTask);
        }

        private void ExecuteTask(TaskWrapperBase TakeTask)
        {
            RuningTasks.TryAdd(TakeTask, default);

            _ = RunExecuteTask(TakeTask);
        }
        private async Task RunExecuteTask(TaskWrapperBase TakeTask)
        {
            try
            {
                await TakeTask.ExecuteTask();
            }
            finally
            {
                if (IsActive)
                {
                    RuningTasks.TryRemove(TakeTask, out _);
                    TryExecuteNextTask();
                }
            }
        }
        private void TaskCompletedEvent(TaskWrapperBase taskWrapper, Task task, object? state)
        {
            RuningTasks?.TryRemove(taskWrapper, out _);
            TryExecuteNextTask();
        }
        private async Task Polling()
        {
            try
            {
                while (IsActive && await PollingTask!.WaitForNextTickAsync(cancellationToken))
                {
                ReDequeue:
                    if (IsActive && RuningTasks.Count < MaximumConcurrencyLevel && !IsCpuLimit && TaskQueue.TryDequeue(out TaskWrapperBase? TakeTask))
                    {
                        if (TakeTask.IsCanceled)
                            goto ReDequeue;
                        ExecuteTask(TakeTask);
                    }
                }
            }
            catch (OperationCanceledException) { }
        }

        public Task Run(Action action, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            TaskWrapper wrapper = CreateTaskWrapper(action, cancellationTokenSource, options);
            QueueTask(wrapper);
            return wrapper.task;
        }

        public Task<TResult> Run<TResult>(Func<TResult> func, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            TaskWrapper<TResult> wrapper = CreateTaskWrapper(func, cancellationTokenSource, options);
            QueueTask(wrapper);
            return wrapper.task;
        }

        public Task<Task<TResult>> Run<TResult>(Func<Task<TResult>> func, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            TaskWrapper<Task<TResult>> wrapper = CreateTaskWrapper(func, cancellationTokenSource, options);
            QueueTask(wrapper);
            return wrapper.task;
        }

        public static TaskWrapper CreateTaskWrapper(Action action, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            cancellationTokenSource ??= new CancellationTokenSource();
            CancellationToken _token = cancellationTokenSource.Token;
            TaskCreationOptions _options = options ?? TaskCreationOptions.None;
            return new TaskWrapper(new Task(action, _token, _options), cancellationTokenSource);
        }

        public static TaskWrapper<TResult> CreateTaskWrapper<TResult>(Func<TResult> func, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            cancellationTokenSource ??= new CancellationTokenSource();
            CancellationToken _token = cancellationTokenSource.Token;
            TaskCreationOptions _options = options ?? TaskCreationOptions.None;
            return new TaskWrapper<TResult>(new Task<TResult>(func, _token, _options), cancellationTokenSource);
        }

        public static TaskWrapper<Task<TResult>> CreateTaskWrapper<TResult>(Func<Task<TResult>> func, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            cancellationTokenSource ??= new CancellationTokenSource();
            CancellationToken _token = cancellationTokenSource.Token;
            TaskCreationOptions _options = options ?? TaskCreationOptions.None;
            return new TaskWrapperTaskResult<TResult>(new Task<Task<TResult>>(func, _token, _options), cancellationTokenSource).TaskWrapper;
        }


        public async Task ParallelForEach<TSource>(IEnumerable<TSource> source, Action<TSource> action, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            List<Task> TaskWrapperList = new(source.Count());
            TaskWrapper wrapper;
            foreach (TSource item in source)
            {
                wrapper = CreateTaskWrapper(() => action(item), cancellationTokenSource, options);
                QueueTask(wrapper);
                TaskWrapperList.Add(wrapper.task);
            }
            await Task.WhenAll(TaskWrapperList).ConfigureAwait(false);
        }
        public async Task<TResult[]> ParallelForEach<TSource, TResult>(IEnumerable<TSource> source, Func<TSource, TResult> func, Action<TResult> CompleteCallback, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            List<Task<TResult>> TaskWrapperList = new(source.Count());
            TaskWrapper<TResult> wrapper;
            foreach (TSource item in source)
            {
                wrapper = CreateTaskWrapper(() => func(item), cancellationTokenSource, options);
                QueueTask(wrapper);
                TaskWrapperList.Add(wrapper.task);
            }
            if (CompleteCallback is not null)
            {
                await foreach (Task<TResult> result in Task.WhenEach(TaskWrapperList).ConfigureAwait(false))
                    CompleteCallback(result.Result);
                return TaskWrapperList.Select(p => p.Result).ToArray();
            }
            await Task.WhenAll(TaskWrapperList).ConfigureAwait(false);
            return TaskWrapperList.Select(p => p.Result).ToArray();
        }
        public async Task<TResult[]> ParallelForEach<TSource, TResult>(IEnumerable<TSource> source, Func<TSource, Task<TResult>> func, Action<TResult> CompleteCallback, CancellationTokenSource? cancellationTokenSource = null, TaskCreationOptions? options = null)
        {
            List<Task<Task<TResult>>> TaskWrapperList = new(source.Count());
            TaskWrapper<Task<TResult>> wrapper;
            foreach (TSource item in source)
            {
                wrapper = CreateTaskWrapper(() => func(item), cancellationTokenSource, options);
                QueueTask(wrapper);
                TaskWrapperList.Add(wrapper.task);
            }
            if (CompleteCallback is not null)
            {
                await foreach (Task<Task<TResult>> result in Task.WhenEach(TaskWrapperList).ConfigureAwait(false))
                    CompleteCallback(result.Result.Result);
                return TaskWrapperList.Select(p => p.Result.Result).ToArray();
            }
            await Task.WhenAll(TaskWrapperList).ConfigureAwait(false);
            return TaskWrapperList.Select(p => p.Result.Result).ToArray();
        }


        #region IDisposable Support
        private volatile bool disposedValue = false;
        protected virtual void OnDispose()
        {
            if (!disposedValue)
            {
                disposedValue = true;
                if (cancellationTokenSource != null)
                {
                    cancellationTokenSource.Cancel();
                    cancellationTokenSource = null;
                }
                PollingTask?.Dispose();
                PollingTask = null;
                // PollingTask?.Dispose();
                // PollingTask = null!;
                if (_cpuPerformanceCount is not null)
                {
                    _cpuPerformanceCount.Dispose();
                    _cpuPerformanceCount = null;
                }
                _cpuPerformanceCountValues = null;
                TaskQueue.Clear();
                RuningTasks.Clear();
                TaskQueueDic.Clear();
            }
        }
        ~LimitedParallelPerformanceTaskManage() { OnDispose(); }
        public void Dispose() { OnDispose(); GC.SuppressFinalize(this); }
        #endregion

    }

}
